package com.unisound.tracelog.server.mq.consumer;

import com.unisound.tracelog.common.model.span.Tag;
import com.unisound.tracelog.common.utils.JsonUtils;
import com.unisound.tracelog.server.model.entity.TraceLogTag;
import com.unisound.tracelog.server.service.TraceLogTagService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

@Slf4j
@Component
public class TraceLogTagKafkaConsumer {

    @Autowired
    private TraceLogTagService traceLogTagService;

    @KafkaListener(topics = {"${tracelog.kafka.consumer.tag.topic}"}, groupId = "${tracelog.kafka.consumer.tag.group-id}")
    public void consume(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        try {
            doConsume(records);
        } catch (Exception e) {
            log.error("TraceLogTag消费异常", e);
        }

        try {
            ack.acknowledge();
        } catch (Exception e) {
            log.error("TraceLogTag提交消费位移异常", e);
        }
    }

    private void doConsume(List<ConsumerRecord<String, String>> records) {
        List<TraceLogTag> traceLogTags = new ArrayList<>();
        for (ConsumerRecord<String, String> record : records) {
            doConsume(record, traceLogTags);
        }

        if (CollectionUtils.isNotEmpty(traceLogTags)) {
            traceLogTagService.saveBatch(traceLogTags);
        }
    }

    private void doConsume(ConsumerRecord<String, String> record, List<TraceLogTag> traceLogTags) {
        List<Tag> tags = JsonUtils.fromJsonList(record.value(), Tag.class);
        if (CollectionUtils.isEmpty(tags)) return;

        tags.forEach(tag -> traceLogTags.add(buildTraceLogTag(tag)));
    }

    private TraceLogTag buildTraceLogTag(Tag tag) {
        return TraceLogTag.builder()
                .id(tag.getId())
                .traceId(tag.getTraceId())
                .tagKey(tag.getTagKey())
                .tagValue(tag.getTagValue())
                .build();
    }

}
